package com.unismc.transport.common;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class DefaultUviewMqttClient implements UviewMqttClient {
    private static final int MSG_QUEUE_SIZE = 65536;
    private static final int CONNECTION_TIMEOUT = 5;
    private final BlockingQueue<UvMsg> msgQueue = new LinkedBlockingDeque<>(MSG_QUEUE_SIZE);
    MqttConnectOptions options;
    @Value("${uv.mqtt.endpoint}")
    private String endpoint;
    @Value("${uv.mqtt.token}")
    private String token;
    @Value("${uv.mqtt.gateway_telemetry_topic:v1/gateway/telemetry}")
    private String gatewayTelemetryTopic;
    @Value("${uv.mqtt.gateway_telemetry_topic:v1/gateway/attributes}")
    private String gatewayAttributesTopic;
    @Value("${uv.mqtt.gateway_connect_topic:v1/gateway/connect}")
    private String gatewayConnectTopic;
    @Value("${uv.mqtt.device_telemetry_topic:v1/devices/me/telemetry}")
    private String deviceTelemetryTopic;
    private String clientId;
    private MqttClient client;

    @PostConstruct
    private void init() {
        clientId = UUID.randomUUID().toString();
        options = new MqttConnectOptions();
        options.setUserName(token);
        options.setConnectionTimeout(CONNECTION_TIMEOUT);
        options.setCleanSession(true);
        options.setKeepAliveInterval(1);
        options.setMaxReconnectDelay(1);

        launchConsumer();
    }

    @Override
    public void createDevice(String deviceName, String type) {
        JSONObject payload = new JSONObject();
        payload.put("device", deviceName);
        payload.put("type", type);

        offer(gatewayConnectTopic, payload.toJSONString());
    }

    @Override
    public void sendAttr(String deviceName, JSONObject msg) {
        JSONObject attributes = new JSONObject();
        attributes.put(deviceName, msg);

        offer(gatewayAttributesTopic, attributes.toJSONString());
    }

    @Override
    public void send(String deviceName, JSONObject msg, Long ts) {
        JSONObject telemetry = new JSONObject();
        telemetry.put("ts", ts);
        telemetry.put("values", msg);
        JSONArray array = new JSONArray();
        array.add(telemetry);
        JSONObject payload = new JSONObject();
        payload.put(deviceName, array);

        offer(gatewayTelemetryTopic, payload.toJSONString());
    }

    @Override
    public void send(String deviceName, JSONObject msg) {
        JSONArray payload = new JSONArray();
        payload.add(msg);

        JSONObject telemetry = new JSONObject();
        telemetry.put(deviceName, payload);

        offer(gatewayTelemetryTopic, telemetry.toJSONString());
    }

    @Override
    public void send(JSONObject msg, Long ts) {
        JSONObject telemetry = new JSONObject();
        telemetry.put("ts", ts);
        telemetry.put("values", msg);

        offer(deviceTelemetryTopic, telemetry.toJSONString());
    }

    @Override
    public void send(JSONObject msg) {
        offer(deviceTelemetryTopic, msg.toJSONString());
    }

    private void offer(String topic, String payload) {
        if (!msgQueue.offer(new UvMsg(topic, payload))) {
            log.info("Msg offer to queue Failed: [{}]", msgQueue.size());
        }
    }

    private void launchConsumer() {
        new Thread(() -> {
            log.info("Mqtt client started!");
            initialize:
            while (!Thread.currentThread().isInterrupted()) {
                do {
                    try {
                        client = new MqttClient(endpoint, clientId, new MemoryPersistence());
                        client.setCallback(mqttCallback);
                        client.connect(options);
                    } catch (MqttException e) {
                        if (e.getCause() instanceof SocketTimeoutException) {
                            log.warn("Mqtt client connect to {}, {}", endpoint, e.getCause().getMessage());
                        } else if (e.getCause() instanceof InterruptedException) {
                            Thread.currentThread().interrupt();
                            continue initialize;
                        } else {
                            log.warn("Mqtt client connect to {}, {}", endpoint, e.getMessage());
                        }
                        try {
                            TimeUnit.SECONDS.sleep(CONNECTION_TIMEOUT);
                        } catch (InterruptedException ex) {
                            Thread.currentThread().interrupt();
                            continue initialize;
                        }
                    }
                } while (!client.isConnected());

                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        UvMsg uvMsg = msgQueue.take();
                        String json = uvMsg.getPayload();
                        if (log.isDebugEnabled()) {
                            log.debug("Mqtt TX data json [{}]", json);
                        }
                        client.publish(uvMsg.getTopic(), json.getBytes(StandardCharsets.UTF_8), 0, false);
//                        log.info("Mqtt TX data length [{}]", json.length());
                        TimeUnit.MILLISECONDS.sleep(10);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    } catch (Exception e) {
                        log.warn("{}", e.getCause() != null ? e.getCause().getMessage() : e.getMessage());
                        try {
                            client.disconnect();
                            client.close();
                        } catch (MqttException ignored) {
                        }

                        continue initialize;
                    }
                }
            }
            log.info("Mqtt client stopped!");
        }).start();
    }

    @Getter
    private static class UvMsg {
        private final String topic;
        private final String payload;
        public UvMsg(String topic, String payload) {
            this.topic = topic;
            this.payload = payload;
        }
    }


    private final MqttCallback mqttCallback = new MqttCallbackExtended() {
        @Override
        public void connectComplete(boolean b, String s) {
            log.info("Mqtt Online, {}", s);
        }

        @Override
        public void connectionLost(Throwable throwable) {
            log.error("Mqtt Offline, {}", throwable.getMessage());
        }

        @Override
        public void messageArrived(String s, MqttMessage mqttMessage) {
            if (log.isTraceEnabled()) {
                log.trace("Mqtt MessageArrived, {}", mqttMessage.toString());
            }
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
            if (log.isTraceEnabled()) {
                log.trace("Mqtt DeliveryComplete, {}", iMqttDeliveryToken.toString());
            }
        }
    };
}
